package com.shengzai.mapreduce.text2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.*;

/**
 * 由于map数据写出时为 Key和Value格式,并且有其类型，所以RecordWriter的泛型可以为
 *      Map输出的类型
 */
public class MyRecordWriter extends RecordWriter<Text, NullWritable> {
    FSDataOutputStream manOutputStream;
    FSDataOutputStream womanOutputStream;
    FileSystem fileSystem;
    OutputStreamWriter manOutputStreamWriter;
    OutputStreamWriter womanOutputStreamWriter;
    BufferedWriter manbufferedWriter;
    BufferedWriter womanbufferedWriter;

    public MyRecordWriter( Configuration configuration) throws IOException {
        // 初始化对象时可以创建FileSystem对象
        fileSystem = FileSystem.get(configuration);
        //  FSDataInputStream fsDataInputStream = fileSystem.open(new Path("/data/emp.csv"));
        //        InputStreamReader inputStreamReader = new InputStreamReader(fsDataInputStream);
        //        BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
        manOutputStream = fileSystem.create(new Path("hadoop_code/src/data/man.txt"));
        womanOutputStream = fileSystem.create(new Path("hadoop_code/src/data/woman.txt"));
        manOutputStreamWriter = new OutputStreamWriter(manOutputStream);
        womanOutputStreamWriter = new OutputStreamWriter(womanOutputStream);
        manbufferedWriter =new BufferedWriter(manOutputStreamWriter);
        womanbufferedWriter = new BufferedWriter(womanOutputStreamWriter);

    }

    public MyRecordWriter() {
    }

    /**
     *  可以在该函数中实现数据写出的功能
     * @param key the key to write.
     * @param value the value to write.
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void write(Text key, NullWritable value) throws IOException, InterruptedException {
        String gander = key.toString().split(",")[3];

        if("男".equals(gander) ){
            manbufferedWriter.write(key.toString()+"\n");
            manbufferedWriter.flush();
        }else{
            womanbufferedWriter.write(key.toString()+"\n");
            womanbufferedWriter.flush();
        }
    }

    /**
     * 该函数主要是用于关闭输出流
     * @param context the context of the task
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {

        manOutputStream.close();
        womanOutputStream.close();
        fileSystem.close();

    }
}
